package com.cml.client.infrastructure;




import com.cml.common.dto.message.FetchAllServerListReponse;
import com.cml.common.dto.message.FetchSlotsAllocationResponse;
import com.cml.common.dto.entity.Message;
import com.cml.common.dto.message.ServiceRegisterResponse;
import com.cml.common.dto.message.ServiceSubscribeResponse;
import org.apache.log4j.Logger;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * 姓名：陈茂林
 * 时间：2023/10/17 9:30
 * 描述：
 */
public class MessageQueueManager {


    public static final Map<String,BlockingQueue<Message>> QUEUE_REQUEST_MANAGER = new HashMap<String, BlockingQueue<Message>>();
    private static final BlockingQueue<Message> RESPONSE_QUEUE_MANAGER = new LinkedBlockingQueue<Message>();
    private static  final Map<String,BlockingQueue<Message>> DATA_EXCHANGE_QUEUE_MANAGER = new HashMap<String, BlockingQueue<Message>>();


    static {
        DispenseThread dispenseThread= new DispenseThread();
        dispenseThread.start();
    }

    public static  void registerDataExchangeQueue(Class classObj,BlockingQueue blockingQueue){
        DATA_EXCHANGE_QUEUE_MANAGER.put(classObj.getName(),blockingQueue);
    }

    public static void removeDataExchangeQueue(Class classObj){
        DATA_EXCHANGE_QUEUE_MANAGER.remove(classObj.getName());
    }
    public static void createQueue(String key){
        QUEUE_REQUEST_MANAGER.put(key, new LinkedBlockingQueue());
    }

    public static void sendRequestMessage(String key,Message message) throws InterruptedException {
        MessageQueueManager.QUEUE_REQUEST_MANAGER.get(key).put(message);
    }


    public static void sendResponseMessage(Message message) throws InterruptedException {
        MessageQueueManager.RESPONSE_QUEUE_MANAGER.put(message);
    }


    public static Message pollRequestMessage(String key){
        return MessageQueueManager.QUEUE_REQUEST_MANAGER.get(key).poll();
    }

    public static Message peekRequestMessage(String key){
        if(MessageQueueManager.QUEUE_REQUEST_MANAGER.get(key) == null){
            return null;
        }
        return MessageQueueManager.QUEUE_REQUEST_MANAGER.get(key).peek();
    }

    public static void removeRequestQueue(String key){
        MessageQueueManager.QUEUE_REQUEST_MANAGER.remove(key);
    }




    private  static  class DispenseThread extends Thread{
        private Logger logger = Logger.getLogger(DispenseThread.class);
        @Override
        public void run() {
            while (true){
                try {
                    Message message = MessageQueueManager.RESPONSE_QUEUE_MANAGER.take();
                    DATA_EXCHANGE_QUEUE_MANAGER.get(message.getClass().getName()).put(message);
                } catch (InterruptedException e) {
                    logger.error(e);
                }
            }
        }
    }
}
